package com.xdemo.processor;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import com.xdemo.handler.MessageActionRouter;
import com.xdemo.msg.IMessage;

public class QueueProcessor<T> implements Runnable,IProcessor<T> {
	
	private BlockingQueue<IMessage<T>> messages = new LinkedBlockingQueue<IMessage<T>>();
	
	public Thread thread  = new Thread(this);
	
	public MessageActionRouter<T> router;
	
	private boolean stoped = false;
	
	public QueueProcessor(MessageActionRouter<T> router) {
		this.router = router;
	}
	
	@Override
	public void execute(IMessage<T> message) {
		messages.add(message);
	}

	@Override
	public void start() {
		thread.start();
	}
	
	@Override
	public void stop() {
		stoped = true;
	}

	@Override
	public void run() {
		while(!stoped) {
			try {
				process(messages.take());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public void process(IMessage<T> message) {
		router.handleMessage(message);
	}

	@Override
	public boolean isFull() {
		return messages.size() > 1000;
	}

}
